package david.java.flink_sql.streamApiIntegration;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

/**
 * @Description:
 * @See https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/data_stream_api/#examples-for-fromchangelogstream
 * @Author: ZhaoDawei
 * @Date: Create in 10:13 上午 2022/4/1
 */
public class S4_1_FromChangeLogStream {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        // === EXAMPLE 1 ===
        // interpret the stream as a retract stream

        // create a changelog DataStream
        DataStreamSource<Row> dataStream = env.fromElements(
                Row.ofKind(RowKind.INSERT, "Alice", 12),
                Row.ofKind(RowKind.INSERT, "Bob", 5),
                Row.ofKind(RowKind.UPDATE_BEFORE, "Alice", 12),
                Row.ofKind(RowKind.UPDATE_AFTER, "Alice", 100)
        );


        Table table = tableEnv.fromChangelogStream(dataStream);

        tableEnv.createTemporaryView("InputTable", table);

        tableEnv.executeSql("select f0, f1 score from InputTable ").print();
        //  print:
        //         +----+--------------------------------+-------------+
        //         | op |                             f0 |       score |
        //         +----+--------------------------------+-------------+
        //         | +I |                          Alice |          12 |
        //         | +I |                            Bob |          5  |
        //         | -U |                          Alice |          12 |
        //         | +U |                          Alice |         100 |
        //         +----+--------------------------------+-------------+


        tableEnv.executeSql("select f0 as name, sum(f1) as score from InputTable group by f0").print();
        // print :
        //         +----+--------------------------------+-------------+
        //         | op |                           name |       score |
        //         +----+--------------------------------+-------------+
        //         | +I |                            Bob |           5 |
        //         | +I |                          Alice |          12 |
        //         | -D |                          Alice |          12 |
        //         | +I |                          Alice |         100 |
        //         +----+--------------------------------+-------------+




        // === EXAMPLE 2 ===

        // interpret the stream as an upsert stream (without a need for UPDATE_BEFORE)

        // create a changelog DataStream

        DataStreamSource<Row> dataStream2 = env.fromElements(
                Row.ofKind(RowKind.INSERT, "Alice", 12),
                Row.ofKind(RowKind.INSERT, "Bob", 5),
                Row.ofKind(RowKind.UPDATE_AFTER, "Alice", 100)
                );

        Table table2 = tableEnv.fromChangelogStream(
                dataStream2,
                Schema.newBuilder().primaryKey("f0").build(),
                ChangelogMode.upsert()
        );

        tableEnv.createTemporaryView("InputTable2", table2);

        tableEnv.executeSql("select f0, f1 from InputTable2").print();
        // print:
        //         +----+--------------------------------+-------------+
        //         | op |                             f0 |          f1 |
        //         +----+--------------------------------+-------------+
        //         | +I |                            Bob |           5 |
        //         | +I |                          Alice |          12 |
        //         | -U |                          Alice |          12 |
        //         | +U |                          Alice |         100 |
        //         +----+--------------------------------+-------------+


        tableEnv.executeSql("SELECT f0 AS name, SUM(f1) AS score FROM InputTable GROUP BY f0").print();
         // print:
         //        +----+--------------------------------+-------------+
         //        | op |                           name |       score |
         //        +----+--------------------------------+-------------+
         //        | +I |                            Bob |           5 |
         //        | +I |                          Alice |          12 |
         //        | -D |                          Alice |          12 |
         //        | +I |                          Alice |         100 |
         //        +----+--------------------------------+-------------+

    }


}
